package com.clown.marketAnalysis

import java.net.URL
import java.sql.Timestamp
import java.time.LocalDate

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

// 定义输入输出样例类
case class AdClickLog(userId: Long, adId: Long, province: String, city: String, timestamp: Long)

case class AdClickCountByProvince(windowEnd: String, province: String, count: Long)

// 侧输出流黑名单报警信息
case class BlackListUserWarning(userId: Long, adId: Long, msg: String)

object AdClickAnalysis {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 读取数据
    val resource: URL = getClass.getResource("/AdClickLog.csv")
    val inputStream: DataStream[String] = env.readTextFile(resource.getPath)
    // 转换成样例类，并提取时间戳和watermark
    val adClickLogStream = inputStream
      .map(data => {
        val arr = data.split(",")
        AdClickLog(arr(0).toLong, arr(1).toLong, arr(2), arr(3), arr(4).toLong)
      })
      .assignAscendingTimestamps(_.timestamp * 1000L)
    // .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[AdClickLog](Time.hours(1)) {
    //   override def extractTimestamp(element: AdClickLog): Long = element.timestamp
    // })

    // 插入一部过滤操作，并将存在刷单行为的用户添加进黑名单报警(侧输出流)
    val tag = new OutputTag[BlackListUserWarning]("warning")
    val maxCount = 100L
    val filterBlackListUserStream: DataStream[AdClickLog] = adClickLogStream
      .keyBy(var2 => (var2.userId, var2.adId))
      .process(new FilterBlackListUserResult(tag, maxCount))

    // 开窗聚合统计
    val adCountResultStream = filterBlackListUserStream
      .keyBy(_.province)
      .timeWindow(Time.hours(1), Time.seconds(5))
      .aggregate(new AdCountAgg(), new AdCountWindowResult())

    adCountResultStream.print("count result")
    filterBlackListUserStream.getSideOutput(tag).print("warning")
    env.execute("ad count statistics job")
  }
}

class AdCountAgg() extends AggregateFunction[AdClickLog, Long, Long] {
  override def createAccumulator(): Long = 0L

  override def add(in: AdClickLog, acc: Long): Long = acc + 1

  override def getResult(acc: Long): Long = acc

  override def merge(acc: Long, acc1: Long): Long = acc + acc1
}


class AdCountWindowResult() extends WindowFunction[Long, AdClickCountByProvince, String, TimeWindow] {
  override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[AdClickCountByProvince]): Unit = {
    out.collect(AdClickCountByProvince(new Timestamp(window.getEnd).toString, key, input.head))
  }
}

class FilterBlackListUserResult(tag: OutputTag[BlackListUserWarning], maxCount: Long) extends KeyedProcessFunction[(Long, Long), AdClickLog, AdClickLog] {
  // 定义状态，保存用户对广告的点击量，每天0点定时清空状态的时间戳
  lazy val countState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("countState", classOf[Long]))
  lazy val resetTimerTimestampState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("resetTimerTimestampState", classOf[Long]))
  lazy val isBlackState: ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("isBlackState", classOf[Boolean]))

  override def processElement(value: AdClickLog, ctx: KeyedProcessFunction[(Long, Long), AdClickLog, AdClickLog]#Context, out: Collector[AdClickLog]): Unit = {
    val currentCount = countState.value()

    // 判断只要是第一个数据来了直接注册0点的清空状态定时器
    if (currentCount == 0) {
      // val timestamp = (ctx.timerService().currentProcessingTime() / (1000 * 60 * 60 * 24) + 1) * (24 * 60 * 60 * 1000) - 8 * 60 * 60 * 1000
      // 获取当天日期+1的凌晨0点时间戳
      val timestamp = Timestamp.valueOf(LocalDate.now.plusDays(1).atStartOfDay).toInstant.toEpochMilli
      resetTimerTimestampState.update(timestamp)
      ctx.timerService().registerProcessingTimeTimer(timestamp)
    }
    // 判断count值是否已经达到定义的阈值，如果超过就输出到黑名单
    if (currentCount >= maxCount) {
      // 判断是否已经在黑名单中，没有的话才输出侧输出流
      if (!isBlackState.value()) {
        isBlackState.update(true)
        ctx.output(tag, BlackListUserWarning(value.userId, value.adId, "Click ad over " + maxCount + " times today."))
      }
      return
    }
    // 正常情况，count加1，然后将数据原样输出
    countState.update(currentCount + 1)
    out.collect(value)
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[(Long, Long), AdClickLog, AdClickLog]#OnTimerContext, out: Collector[AdClickLog]): Unit = {
    if (timestamp == resetTimerTimestampState.value()) {
      isBlackState.clear()
      countState.clear()
    }
  }
}